package com.sailing.lianxi.kafka;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import com.sailing.lianxi.common.Constants;
import com.sailing.lianxi.common.ObjTool;
import com.sailing.lianxi.entity.UserInfo;
import com.sailing.lianxi.service.MessageDao;
import com.sailing.lianxi.service.MyConsumerRebalancerListener;
import com.sailing.lianxi.service.ProcessDataService;

public class MyConsumer{

	static String TOPIC =Constants.TOPIC_NAME;
	/**
	 * 自动偏移提交
	 */
	public  static void autoOffsetSubmit(){
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
		props.put("client.id", "consumer1"); 
		//kafka组
        props.put("group.id", "1"); 
        /*
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
                 以上配置只对一个组里没有消费的消息起作用
        */
//        props.put("auto.offset.reset", "earliest"); 
        //自动提交offset--自动提交模式
        props.put("enable.auto.commit", "true");  
        //每隔1s，自动提交offsets-自动提交模式
        props.put("auto.commit.interval.ms", "1000");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程
        props.put("session.timeout.ms", "30000");  
        //key序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
        //值序列化器
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 * 1024*100);//设置最大拉取SIZE
	     //创建消费者
	     KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
	     //订阅主题列表topic
	     consumer.subscribe(Arrays.asList(TOPIC));
	     //指定消费者消费0号分区
//	     consumer.assign(Arrays.asList(new TopicPartition(TOPIC, 0)));
	     while(true){
	    	 ConsumerRecords<String, String> records = consumer.poll(1000);//1秒获取一次
	    	 for (ConsumerRecord<String, String> record : records) {
	    		//　正常这里应该使用线程池处理，不应该在这里处理
	    		 System.out.printf("partition=%d,offset=%d,key=%s,value=%s",record.partition(),record.offset(),record.key(),record.value()+"\n");
			}
	    	 
	     }
	}
	
	/**
	 * 手动偏移控制
	 */
	public  static void manualOffsetControl(){
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
		//kafka组
        props.put("group.id", "wg10");  
        /*
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
                 以上配置只对一个组里没有消费的消息起作用
        */
        props.put("auto.offset.reset", "earliest");  
        props.put("enable.auto.commit", "false");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程,默认6秒
        props.put("session.timeout.ms", "6000");//毫秒 
        //key序列化器
        String deserialClass = StringDeserializer.class.getName();
        props.put("key.deserializer", deserialClass);  
        //值序列化器
        props.put("value.deserializer", deserialClass);
        
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 *1024*100);//设置最大拉取SIZE
        props.put("max.poll.records", 1);
	     //创建消费者
	     KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
	     //订阅主题列表topic
	     consumer.subscribe(Arrays.asList(TOPIC));
	     //最小提交记录数
	     while(true){
	    	 ConsumerRecords<String, String> records = consumer.poll(1000);//1秒获取一次
//	    	 int count = records.count();
//	    	 System.out.println("一次拉去的记录条数:"+count);
	    	 for (ConsumerRecord<String, String> record : records) {
	    		//　正常这里应该使用线程池处理，不应该在这里处理
	    		 System.out.printf("partition=%d,offset=%d,key=%s,value=%s",record.partition(),record.offset(),record.key(),record.value()+"\n");
			  }
//	    	 try {
//	    		 //添加等待可以测试超时问题
//	    	     System.out.println("停止10秒"+new Date());
//                  Thread.sleep(10*1000);
//            } catch (InterruptedException e) {
//                // TODO Auto-generated catch block
//                e.printStackTrace();
//            }
	    	 //手动提交
	    	 consumer.commitSync();
	     }
	}
	
	
	/**
     * 消费byte数据
     */
    public  static void consumerByteData(){
        Properties props = new Properties();
        //kafka服务器地址  
        props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
        //kafka组
        props.put("group.id", "wg_gp");  
        /*
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
                 以上配置只对一个组里没有消费的消息起作用
        */
        props.put("auto.offset.reset", "earliest");  
        props.put("enable.auto.commit", "false");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程
        props.put("session.timeout.ms", "30000");  
        props.put("heartbeat.interval.ms", 10000);
        props.put("request.timeout.ms", 31000);
        //key序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");  
        //值序列化器
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 * 1024*100);//设置最大拉取SIZE
         //创建消费者
         KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props);
         //订阅主题列表topic
         consumer.subscribe(Arrays.asList(TOPIC));
         //最小提交记录数
         while(true){
             ConsumerRecords<byte[], byte[]> records = consumer.poll(1000);//1秒获取一次
             for (ConsumerRecord<byte[], byte[]> record : records) {
                //　正常这里应该使用线程池处理，不应该在这里处理
//                 System.out.printf("partition=%d,offset=%d,key=%s,value=%s",record.partition(),record.offset(),record.key(),record.value()+"\n");
                UserInfo userInfo = (UserInfo) ObjTool.ByteToObject(record.value());
                System.out.println("消费者:"+userInfo.getUuid()+",userName="+userInfo.getUserName());
             }
             //手动提交
             consumer.commitSync();
             
         }
    }
	public static  void getPartitionsFor(){
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
		//kafka组
        props.put("group.id", "4");  
       
        props.put("auto.offset.reset", "earliest"); 
        //自动提交offset
        props.put("enable.auto.commit", "true");  
        //每隔1s，自动提交offsets
        props.put("auto.commit.interval.ms", "1000");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程
        props.put("session.timeout.ms", "30000");  
        //key序列化器
        props.put("key.deserializer", StringDeserializer.class.getName());  
        //值序列化器
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 * 1024*100);//设置最大拉取SIZE
	     //创建消费者
	    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
	    
	    List<TopicPartition> partitions = new ArrayList<TopicPartition>();
		  for (PartitionInfo partition : consumer.partitionsFor(TOPIC)){
			  System.out.println(partition.leader()); 
			  partitions.add(new TopicPartition(TOPIC, partition.partition()));
		       consumer.assign(partitions); 
		  }
	}
	
	
	/**
	 * 只消费一次
	 */
	public static void exactlyOnce() {
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
		//kafka组
        props.put("group.id", "wg10");  
        /*
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
                 以上配置只对一个组里没有消费的消息起作用
        */
        props.put("auto.offset.reset", "earliest");  
        props.put("enable.auto.commit", "false");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程,默认6秒
        props.put("session.timeout.ms", "6000");//毫秒 
        //key序列化器
        String deserialClass = StringDeserializer.class.getName();
        props.put("key.deserializer", deserialClass);  
        //值序列化器
        props.put("value.deserializer", deserialClass);
        
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 *1024*100);//设置最大拉取SIZE
        props.put("max.poll.records", 1);
	     //创建消费者
	     KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
	     MessageDao messageDao = new MessageDao();
	     ProcessDataService service = new ProcessDataService();
	     MyConsumerRebalancerListener rebalancerListener = new MyConsumerRebalancerListener(consumer,messageDao);
	     //订阅主题列表topic
	     consumer.subscribe(Arrays.asList(TOPIC),rebalancerListener);
	     //最小提交记录数
	     while(true){
	    	 ConsumerRecords<String, String> records = consumer.poll(100);//1秒获取一次
	    	 for (ConsumerRecord<String, String> record : records) {
	    		  boolean isException=false;
	    		  try {
	    			  service.process(record);
				} catch (Exception e) {
					isException = true;
					e.printStackTrace();
				}
	    		if(isException) {
	    			 //处理发生异常，说明数据没有被消费，为了保证能被消费，需要移动到该位置继续进行处理
	    			TopicPartition topicPartition = new TopicPartition(record.topic(),record.partition());
	    			consumer.seek(topicPartition, record.offset());
	    			System.out.printf("coxnsume exception offset:%d",record.offset());
	    		}
			  }
	     }
	}
	
	
	
	
	public static void main(String[] args) {
		manualOffsetControl();
	}
}
